/*
 *
 * Copyright (c) 2020 The Raptor Authors. All rights reserved.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 *
 */

#include "src/windows/tcp_listener.h"
#include <string.h>

#include "raptor-lite/impl/acceptor.h"
#include "raptor-lite/impl/config.h"
#include "raptor-lite/impl/endpoint.h"
#include "raptor-lite/impl/property.h"
#include "raptor-lite/utils/list_entry.h"
#include "raptor-lite/utils/log.h"

#include "src/common/endpoint_impl.h"
#include "src/common/socket_util.h"
#include "src/windows/socket_settings.h"
#include "src/windows/iocp_thread.h"

namespace raptor {

struct TcpListenObject {
    ListEntry entry;
    SOCKET new_socket;
    int listen_port;
    raptor_dualstack_mode mode;
    uint8_t addr_buffer[(sizeof(raptor_sockaddr_in6) + 16) * 2];
    OverLappedEx olex;
};

TcpListener::TcpListener(AcceptorHandler *service)
    : service_(service)
    , shutdown_(true)
    , AcceptEx_(nullptr)
    , GetAcceptExSockAddrs_(nullptr)
    , poll_thread_(nullptr) {
    RAPTOR_LIST_INIT(&list_head_);
}

TcpListener::~TcpListener() { Shutdown(); }

raptor_error TcpListener::Init(PollingThread *poll_thread) {
    if (!shutdown_) return RAPTOR_ERROR_FROM_STATIC_STRING("TcpListener has been initialized");

    bool success = false;
    thd_ = Thread("TcpListener", std::bind(&TcpListener::WorkThread, this, std::placeholders::_1),
                  nullptr, &success);
    if (!success) {
        return RAPTOR_ERROR_FROM_STATIC_STRING("TcpListener: Failed to create thread");
    }

    shutdown_ = false;
    poll_thread_ = poll_thread;
    count_.Store(0);
    return RAPTOR_ERROR_NONE;
}

raptor_error TcpListener::Start() {
    if (shutdown_) {
        return RAPTOR_ERROR_FROM_STATIC_STRING("TcpListener has not been initialized");
    }

    thd_.Start();
    return RAPTOR_ERROR_NONE;
}

void TcpListener::Shutdown() {
    if (!shutdown_) {
        log_warn("TcpListener: prepare to shutdown");
        shutdown_ = true;
        cv_.Signal();
        thd_.Join();

        mtx_.Lock();
        // clear async listen object
        ListEntry *entry = nullptr;
        do {
            entry = ListPopBack(&list_head_);
            if (entry) {
                auto obj = reinterpret_cast<TcpListenObject *>(entry);
                if (obj->new_socket != INVALID_SOCKET) {
                    closesocket(obj->new_socket);
                }
                delete obj;
            }
        } while (entry);

        for (int i = 0; i < static_cast<int>(listen_fds_.size()); i++) {
            poll_thread_->Delete(listen_fds_[i]);
            closesocket(listen_fds_[i]);
        }
        listen_fds_.clear();
        mtx_.Lock();

        // clear message queue
        bool empty = true;
        do {
            auto n = mpscq_.PopAndCheckEnd(&empty);
            auto msg = reinterpret_cast<PollingThreadMessageNode *>(n);
            if (msg != nullptr) {
                delete msg;
            }
        } while (!empty);
        log_warn("TcpListener: shutdown");
    }
}

raptor_error TcpListener::AddListeningPort(const raptor_resolved_address *addr) {
    if (shutdown_) return RAPTOR_ERROR_FROM_STATIC_STRING("TcpListener has been shutdown");
    raptor_resolved_address mapped_addr;
    raptor_dualstack_mode mode;
    SOCKET listen_fd;

    raptor_error e = raptor_create_socket(addr, &mapped_addr, &listen_fd, &mode);

    if (e != RAPTOR_ERROR_NONE) {
        log_error("TcpListener: Failed to create socket: %s", e->ToString().c_str());
        return e;
    }

    if (!AcceptEx_ || !GetAcceptExSockAddrs_) {
        e = GetExtensionFunction(listen_fd);
        if (e != RAPTOR_ERROR_NONE) {
            closesocket(listen_fd);
            return e;
        }
    }

    int port = 0;
    e = raptor_tcp_server_prepare_socket(listen_fd, &mapped_addr, &port, 1);
    if (e != RAPTOR_ERROR_NONE) {
        log_error("TcpListener: Failed to configure socket: %s", e->ToString().c_str());
        return e;
    }

    mtx_.Lock();
    listen_fds_.push_back(listen_fd);
    poll_thread_->Add(listen_fd, this);
    mtx_.Unlock();

    struct TcpListenObject *node = new TcpListenObject;
    RAPTOR_LIST_ENTRY_INIT(&node->entry);
    node->new_socket = INVALID_SOCKET;
    node->listen_port = port;
    node->mode = mode;
    memset(node->addr_buffer, 0, sizeof(node->addr_buffer));
    memset(&node->olex, 0, sizeof(node->olex));
    node->olex.event_type = internal::kAcceptEvent;
    node->olex.user_id = 0;

    ListPushBack(&list_head_, &node->entry);

    e = StartAcceptEx(listen_fd, &mapped_addr, node);
    if (e != RAPTOR_ERROR_NONE) {
        log_error("TcpListener: Failed to StartAcceptEx, %s", e->ToString().c_str());
        ListRemoveEntry(&node->entry);
        delete node;
        return e;
    }

    char *addr_string = nullptr;
    raptor_sockaddr_to_string(&addr_string, &mapped_addr, 0);
    log_info("TcpListener: start listening on %s",
             addr_string ? addr_string : std::to_string(port).c_str());
    if (addr_string) free(addr_string);

    return RAPTOR_ERROR_NONE;
}

raptor_error TcpListener::GetExtensionFunction(SOCKET fd) {

    DWORD NumberofBytes;
    int status = 0;

    if (!AcceptEx_) {
        GUID guid = WSAID_ACCEPTEX;
        status = WSAIoctl(fd, SIO_GET_EXTENSION_FUNCTION_POINTER, &guid, sizeof(guid), &AcceptEx_,
                          sizeof(AcceptEx_), &NumberofBytes, NULL, NULL);

        if (status != 0) {
            AcceptEx_ = NULL;
            raptor_error e = RAPTOR_WINDOWS_ERROR(WSAGetLastError(), "WSAIoctl");
            log_error("TcpListener: Failed to get AcceptEx: %s", e->ToString().c_str());
            return e;
        }
    }

    if (!GetAcceptExSockAddrs_) {
        GUID guid = WSAID_GETACCEPTEXSOCKADDRS;
        status = WSAIoctl(fd, SIO_GET_EXTENSION_FUNCTION_POINTER, &guid, sizeof(guid),
                          &GetAcceptExSockAddrs_, sizeof(GetAcceptExSockAddrs_), &NumberofBytes,
                          NULL, NULL);

        if (status != 0) {
            GetAcceptExSockAddrs_ = NULL;
            raptor_error e = RAPTOR_WINDOWS_ERROR(WSAGetLastError(), "WSAIoctl");
            log_error("TcpListener: Failed to get GetAcceptexSockAddrs: %s", e->ToString().c_str());
            return e;
        }
    }
    return RAPTOR_ERROR_NONE;
}

void TcpListener::OnEventProcessImpl(EventDetail *detail) {
    size_t offset = offsetof(TcpListenObject, olex);
    TcpListenObject *object = reinterpret_cast<TcpListenObject *>(
        reinterpret_cast<intptr_t>(detail->overlaped) - static_cast<intptr_t>(offset));

    SOCKET listen_fd = reinterpret_cast<SOCKET>(detail->ptr);

    raptor_resolved_address client, server;
    memset(&client, 0, sizeof(client));
    memset(&server, 0, sizeof(server));
    ParsingNewConnectionAddress(object, &server, &client);

    auto ep = std::make_shared<EndpointImpl>(object->new_socket, &server, &client);
    ep->SetListenPort(static_cast<uint16_t>(object->listen_port));

    Property property;
    service_->OnAccept(ep, property);
    if (ep->IsValid() && property.Count() > 0) {
        ProcessProperty(object->new_socket, property);
    }

    object->new_socket = INVALID_SOCKET;
    raptor_error e = StartAcceptEx(listen_fd, &server, object);

    if (e != RAPTOR_ERROR_NONE) {
        log_error("TcpListener: Failed to StartAcceptEx for next fd, %s", e->ToString().c_str());
        Shutdown();
    }
}

raptor_error TcpListener::StartAcceptEx(SOCKET listen_fd, raptor_resolved_address *addr,
                                        struct TcpListenObject *sp) {
    BOOL success = false;
    DWORD addrlen = sizeof(raptor_sockaddr_in6) + 16;
    DWORD bytes_received = 0;
    raptor_error error = RAPTOR_ERROR_NONE;
    SOCKET sock;

    sock = WSASocket(((raptor_sockaddr *)addr->addr)->sa_family, SOCK_STREAM, IPPROTO_TCP, NULL, 0,
                     RAPTOR_WSA_SOCKET_FLAGS);

    if (sock == INVALID_SOCKET) {
        error = RAPTOR_WINDOWS_ERROR(WSAGetLastError(), "WSASocket");
        goto failure;
    }

    error = raptor_tcp_prepare_socket(sock, 0);
    if (error != RAPTOR_ERROR_NONE) goto failure;

    sp->new_socket = sock;

    /* Start the "accept" asynchronously. */
    success = AcceptEx_(listen_fd, sock, sp->addr_buffer, 0, addrlen, addrlen, &bytes_received,
                        &sp->olex.overlapped);

    /* It is possible to get an accept immediately without delay. However, we
        will still get an IOCP notification for it. So let's just ignore it. */
    if (!success) {
        int last_error = WSAGetLastError();
        if (last_error != ERROR_IO_PENDING) {
            error = RAPTOR_WINDOWS_ERROR(WSAGetLastError(), "AcceptEx");
            goto failure;
        }
    }

    // We're ready to do the accept.
    return RAPTOR_ERROR_NONE;

failure:
    if (sock != INVALID_SOCKET) {
        closesocket(sock);
    }
    sp->new_socket = INVALID_SOCKET;
    return error;
}

void TcpListener::ParsingNewConnectionAddress(const TcpListenObject *sp,
                                              raptor_resolved_address *server,
                                              raptor_resolved_address *client) {

    raptor_sockaddr *local = NULL;
    raptor_sockaddr *remote = NULL;

    int local_addr_len = sizeof(raptor_sockaddr_in6) + 16;
    int remote_addr_len = sizeof(raptor_sockaddr_in6) + 16;

    GetAcceptExSockAddrs_((void *)sp->addr_buffer, 0, sizeof(raptor_sockaddr_in6) + 16,
                          sizeof(raptor_sockaddr_in6) + 16, &local, &local_addr_len, &remote,
                          &remote_addr_len);

    if (local != nullptr) {
        server->len = local_addr_len;
        memcpy(server->addr, local, local_addr_len);
    }

    if (remote != nullptr) {
        client->len = remote_addr_len;
        memcpy(client->addr, remote, remote_addr_len);
    }
}

void TcpListener::ProcessProperty(SOCKET fd, const Property &p) {
    bool SocketNoSIGPIPE = true;
    if (p.CheckValue<bool>(config::kSocketNoSIGPIPE, SocketNoSIGPIPE) && SocketNoSIGPIPE) {
        raptor_set_socket_no_sigpipe_if_possible(fd);
    }

    bool SocketReuseAddress = true;
    if (p.CheckValue<bool>(config::kSocketReuseAddress, SocketReuseAddress) &&
        !SocketReuseAddress) {
        raptor_set_socket_reuse_addr(fd, 0);
    }

    bool SocketReusePort = true;
    if (p.CheckValue<bool>(config::kSocketReusePort, SocketReusePort) && !SocketReusePort) {
        raptor_set_socket_reuse_port(fd, 0);
    }

    bool SocketLowLatency = true;
    if (p.CheckValue<bool>(config::kSocketLowLatency, SocketLowLatency) && !SocketLowLatency) {
        raptor_set_socket_low_latency(fd, 0);
    }

    int SocketSendTimeoutMs = 0;
    if (p.CheckValue<int>(config::kSocketSendTimeoutMs, SocketSendTimeoutMs) &&
        SocketSendTimeoutMs > 0) {
        raptor_set_socket_snd_timeout(fd, SocketSendTimeoutMs);
    }

    int SocketRecvTimeoutMs = 0;
    if (p.CheckValue<int>(config::kSocketRecvTimeoutMs, SocketRecvTimeoutMs) &&
        SocketRecvTimeoutMs > 0) {
        raptor_set_socket_rcv_timeout(fd, SocketRecvTimeoutMs);
    }

    bool SocketNonBlocking = true;
    if (p.CheckValue<bool>(config::kSocketNonBlocking, SocketNonBlocking) && !SocketNonBlocking) {
        raptor_set_socket_nonblocking(fd, 0);
    }
}

void TcpListener::OnEventProcess(EventDetail *detail) {
    if (shutdown_) return;
    auto msg = new PollingThreadMessageNode;
    memcpy(&msg->detail, detail, sizeof(*detail));
    mpscq_.push(&msg->node);
    count_.FetchAdd(1, MemoryOrder::RELAXED);
    cv_.Signal();
}

void TcpListener::WorkThread(void *) {
    while (!shutdown_) {
        if (count_.Load() < 1) {
            AutoMutex g(&cmtx_);
            cv_.Wait(&cmtx_, 1000);
            continue;
        }
        auto n = mpscq_.pop();
        auto msg = reinterpret_cast<PollingThreadMessageNode *>(n);
        if (msg != nullptr) {
            count_.FetchSub(1, MemoryOrder::RELAXED);
            this->OnEventProcessImpl(&msg->detail);
            delete msg;
        }
    }
}
} // namespace raptor
